/*
Code : Dzlua
Email: 505544956@qq.com
Time : 2017/06/05
*/
#include "multi_server.h"

namespace dzlua {

namespace cs {

Conn::Conn(int fd)
        : m_fd(fd)
        , m_ReadBuf(nullptr)
        , m_WriteBuf(nullptr)
        , m_Prev(nullptr)
        , m_Next(nullptr)
        , m_Thread(nullptr) {}

Conn::~Conn() {}

LibeventThread *Conn::GetThread() {
    return m_Thread;
}

int Conn::GetFd() {
    return m_fd;
}

int Conn::GetReadBufferLen() {
    return evbuffer_get_length(m_ReadBuf);
}

int Conn::GetReadBuffer(char *buffer, int len) {
    return evbuffer_remove(m_ReadBuf, buffer, len);
}

int Conn::CopyReadBuffer(char *buffer, int len) {
    return evbuffer_copyout(m_ReadBuf, buffer, len);
}

int Conn::GetWriteBufferLen() {
    return evbuffer_get_length(m_WriteBuf);
}

int Conn::AddToWriteBuffer(char *buffer, int len) {
    return evbuffer_add(m_WriteBuf, buffer, len);
}

void Conn::MoveBufferData() {
    evbuffer_add_buffer(m_WriteBuf, m_ReadBuf);
}

//--------------------------//
ConnQueue::ConnQueue() {
    /*
        Establish the head and tail node
            and adjust its pointer.
    */
    m_head = new Conn(0);
    m_tail = new Conn(0);
    m_head->m_Prev = m_tail->m_Next = nullptr;
    m_head->m_Next = m_tail;
    m_tail->m_Prev = m_head;
}

ConnQueue::~ConnQueue() {
    Conn *tcur, *tnext;
	tcur = m_head;
	// Delete each node in the chain list.
    while( tcur != nullptr ) {
		tnext = tcur->m_Next;
		delete tcur;
		tcur = tnext;
	}
}

Conn *ConnQueue::InsertConn(int fd, LibeventThread *t) {
    Conn *c = new Conn(fd);
	c->m_Thread = t;
	Conn *next = m_head->m_Next;

	c->m_Prev = m_head;
	c->m_Next = m_head->m_Next;
	m_head->m_Next = c;
	next->m_Prev = c;
	return c;
}

void ConnQueue::DeleteConn(Conn *c) {
    c->m_Prev->m_Next = c->m_Next;
	c->m_Next->m_Prev = c->m_Prev;
	delete c;
}

//--------------------------//
MultiServer::MultiServer(int count)
        : m_ThreadCount(count)
        , m_Port(-1)
        , m_MainBase(nullptr)
        , m_Threads(nullptr) {
			
#if defined(WIN32)
	{
		WORD wVersion;
		WSADATA wsaData;
		wVersion = MAKEWORD(2, 2);
		int err = WSAStartup(wVersion, &wsaData);
		if (0 != err) {
			printf("error init.\n");
			exit(1);
		}

		if (wsaData.wVersion != wVersion) {
			printf("error init\n");
			WSACleanup();
			exit(1);
		}
	}
#endif

	evthread_use_windows_threads();

    m_MainBase = new LibeventThread;
    m_Threads = new LibeventThread[m_ThreadCount];
    m_MainBase->td = nullptr;
    m_MainBase->base = event_base_new();
    memset(m_SignalEvents, 0, sizeof(m_SignalEvents));

    for(int i = 0; i < m_ThreadCount; ++i) {
		SetupThread(&m_Threads[i]);
	}
}

MultiServer::~MultiServer() {
    StopRun(nullptr);

    event_base_free(m_MainBase->base);
    
    for(int i = 0; i < m_ThreadCount; ++i){
        event_base_free(m_Threads[i].base);
    }

    delete m_MainBase;
    delete [] m_Threads;

#if defined(WIN32)
	WSACleanup();
#endif

}

void MultiServer::SetPort(int port) {
    m_Port = port;
}

bool MultiServer::StartRun() {
    evconnlistener *listener = nullptr;

    /*
        If the port number is not EXIT_CODE,
            listen to the port number.
    */
    if( m_Port != EXIT_CODE ) {
        sockaddr_in sin;
		memset(&sin, 0, sizeof(sin));
		sin.sin_family = AF_INET;
		sin.sin_port = htons(m_Port);
		listener = evconnlistener_new_bind(
                        m_MainBase->base,
                        ListenerEventCb, (void*)this,
                        LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE, -1,
                        (sockaddr*)&sin, sizeof(sockaddr_in));
		if( nullptr == listener ) {
			fprintf(stderr, "listen error: %s\n", strerror(errno));
			exit(1);
		}
    }

    // start each child thread.
    for(int i = 0; i < m_ThreadCount; ++i) {
		m_Threads[i].td.reset(new std::thread(
				&MultiServer::WorkerLibevent,
				(void*)&m_Threads[i] ));
    }

	printf("listening...\n");
    // start The event loop of the primary thread.
    event_base_dispatch(m_MainBase->base);

    // Event loop done, release listener memory.
    if( m_Port != EXIT_CODE ) {
        evconnlistener_free(listener);
    }
}

void MultiServer::StopRun(timeval *tv) {
    int contant = EXIT_CODE;

    /*
        Write EXIT_CODE to the management of the sub threads
            and notify them to exit.
    */
    for(int i = 0; i < m_ThreadCount; ++i) {  
		send(m_Threads[i].notifySendFd, (char*)&contant, sizeof(evutil_socket_t), 0);
	}

    // stop the event loop of the main thread.
    event_base_loopexit(m_MainBase->base, tv);
}

bool MultiServer::AddSignalEvent(int sig, void (*ptr)(int, short, void*)) {
    if( sig >= MAX_SIGNAL )
		return false;

    // Create a new signal event.
	event *ev = evsignal_new(m_MainBase->base, sig, ptr, (void*)this);
	if ( !ev || event_add(ev, nullptr) < 0 ) {
		event_del(ev);
		return false;
	}

	/*
        Delete an old signal event.
        The same signal can only have one signal event.
    */
	if( nullptr != m_SignalEvents[sig] )
		DeleteSignalEvent(sig);
	m_SignalEvents[sig] = ev;

	return true;
}

bool MultiServer::DeleteSignalEvent(int sig) {
    event *ev = m_SignalEvents[sig];

	if( sig >= MAX_SIGNAL || nullptr == ev )
		return false;

	event_del(ev);
	ev = nullptr;

	return true;
}

event *MultiServer::AddTimerEvent(void(*ptr)(int, short, void*), timeval tv, bool once) {
    int flag = 0;
	if( !once ) flag = EV_PERSIST;

	// Create a new timer signal event.
	event *ev = new event;
	event_assign(ev, m_MainBase->base, -1, flag, ptr, (void*)this);
	if( event_add(ev, &tv) < 0 ) {
		event_del(ev);
		return nullptr;
	}

	return ev;
}

bool MultiServer::DeleteTImerEvent(event *ev) {
    int res = event_del(ev);
	return (0 == res);
}

void MultiServer::SetupThread(LibeventThread *ltd) {
	// Establish libevent event handling mechanism.
	ltd->tcpConnect = this;

#ifdef WIN32
        evutil_socket_t fds[2];
        if(evutil_socketpair(AF_INET, SOCK_STREAM, 0, fds) < 0) {
            perror("Error : SetupThread:: evutil_socketpair.\n");
            exit(1);
        }
        evutil_make_socket_nonblocking(fds[0]);
        evutil_make_socket_nonblocking(fds[1]);
#else
        int fds[2];
        if (pipe(fds) < 0) {
            perror("Error : SetupThread:: Can't create notify pipe.\n");
            exit(1);
        }
#endif

	ltd->notifyReceiveFd = fds[0];
	ltd->notifySendFd 	 = fds[1];

	ltd->base = event_base_new();
	if (ltd->base == nullptr) {
		perror("Error : SetupThread:: event_base_new.\n");
		exit(1);
	}

	// Let the child thread's state machine listen to the pipe.
	event_set( &ltd->notifyEvent, ltd->notifyReceiveFd,
		    EV_READ | EV_PERSIST, &MultiServer::ThreadProcess, ltd );
	event_base_set(ltd->base, &ltd->notifyEvent);
	if ( event_add(&ltd->notifyEvent, nullptr) < 0) {
		perror("Error : SetupThread:: event_add.\n");
		exit(1);
	}
	
}

// static 
void MultiServer::WorkerLibevent(void *arg) {
    /*
        Turn on the libevent event loop,
            ready to process business.
    */
	LibeventThread *ltd = (LibeventThread*)arg;
	event_base_dispatch(ltd->base);
}

// static 
void MultiServer::ThreadProcess(int fd, short which, void *arg) {
    LibeventThread *ltd = (LibeventThread*)arg;

	/*
        Read data from a pipe.
        Descriptor or operation code of socket.
    */
	int pipefd = ltd->notifyReceiveFd;
	evutil_socket_t confd;
	recv(pipefd, (char*)&confd, sizeof(evutil_socket_t), 0);

	/*
        If the opcode is EXIT_CODE, 
            the event loop will be break.
    */
	if( EXIT_CODE == confd ) {
		event_base_loopbreak(ltd->base);
		return;
	}

	// make new connection.
	struct bufferevent *bev = nullptr;
	bev = bufferevent_socket_new(ltd->base, confd, BEV_OPT_CLOSE_ON_FREE);
	if (!bev) {
		fprintf(stderr, "Error constructing bufferevent!");
		event_base_loopbreak(ltd->base);
		return;
	}

	// Put the link in the queue.
	Conn *conn = ltd->connectQueue.InsertConn(confd, ltd);

	// Ready to read and write data from socket
	bufferevent_setcb(bev,
			&MultiServer::ReadEventCb,
			&MultiServer::WriteEventCb,
			&MultiServer::CloseEventCb, conn);
	bufferevent_enable(bev, EV_WRITE);
	bufferevent_enable(bev, EV_READ);

	// Call user-defined connection event handler.
	ltd->tcpConnect->ConnectionEvent(conn);
}

// static 
void MultiServer::ListenerEventCb(struct evconnlistener *listener
            , evutil_socket_t fd, sockaddr *sa
            , int socklen, void *user_data) {
    MultiServer *server = (MultiServer*)user_data;

	/*
        Randomly select a child thread, 
            and pass the socket descriptor through the pipe.
    */
	int num = rand() % server->m_ThreadCount;
	int sendfd = server->m_Threads[num].notifySendFd;
	send(sendfd, (char*)&fd, sizeof(evutil_socket_t), 0);
}

// static 
void MultiServer::ReadEventCb(struct bufferevent *bev, void *data) {
    Conn *conn = (Conn*)data;
	conn->m_ReadBuf = bufferevent_get_input(bev);
	conn->m_WriteBuf = bufferevent_get_output(bev);

	// Call user-defined read event handler function.
	conn->m_Thread->tcpConnect->ReadEvent(conn);
}

// static 
void MultiServer::WriteEventCb(struct bufferevent *bev, void *data) {
    Conn *conn = (Conn*)data;
	conn->m_ReadBuf = bufferevent_get_input(bev);
	conn->m_WriteBuf = bufferevent_get_output(bev);

	// Call user-defined write event handler functions.
	conn->m_Thread->tcpConnect->WriteEvent(conn);
}

// static 
void MultiServer::CloseEventCb(struct bufferevent *bev, short events, void *data) {
    Conn *conn = (Conn*)data;
	// Call user-defined break event handler function.
	conn->m_Thread->tcpConnect->CloseEvent(conn, events);
	conn->GetThread()->connectQueue.DeleteConn(conn);
	bufferevent_free(bev);
}

} // end namespace cs

} // end namespace dzlua